package com.chb.flink.table


import com.chb.flink.source.{MyCustomerSource, StationLog}
import org.apache.flink.api.scala.typeutils.Types
import org.apache.flink.streaming.api.scala.{DataStream, StreamExecutionEnvironment}
import org.apache.flink.table.api.scala.StreamTableEnvironment
import org.apache.flink.table.api.{EnvironmentSettings, Table}
import org.apache.flink.table.sources.CsvTableSource
import org.apache.flink.types.Row

object TestCreateTableByDataStream {
    //使用Flink原生的代码创建TableEnvironment
    //先初始化流计算的上下文
    val streamEnv: StreamExecutionEnvironment = StreamExecutionEnvironment.getExecutionEnvironment
    val settings: EnvironmentSettings = EnvironmentSettings.newInstance().useOldPlanner().inStreamingMode().build()
    val tableEnv: StreamTableEnvironment = StreamTableEnvironment.create(streamEnv, settings)


    import org.apache.flink.streaming.api.scala._ // 隐式转换

    import org.apache.flink.table.api.scala._ // 第二个隐式转换

    def main(args: Array[String]): Unit = {
        val dataStream = streamEnv.addSource(new MyCustomerSource)
        // creatDynamicTable(dataStream)
        // modifyFields(dataStream)
        // filterAndSelect(dataStream)
        groupAndAggregate(dataStream)


    }

    /**
     * 修改字段
     */
    def modifyFields(dataStream: DataStream[StationLog]): Unit = {
        val table = tableEnv.fromDataStream(dataStream, 's_id, 'call_out, 'call_in, 'call_type)
        table.printSchema()
    }

    /**
     * 过滤查询
     *
     * @param dataStream
     */
    def filterAndSelect(dataStream: DataStream[StationLog]) = {
        val table = tableEnv.fromDataStream(dataStream)
        tableEnv.toAppendStream[Row](
            table.filter('callType === "success") //filter
                .where('callType === "success")) //where
            .print()

        tableEnv.execute("cs")
    }

    /**
     * 分组聚合
     */
    def groupAndAggregate(dataStream: DataStream[StationLog]) = {
        val table = tableEnv.fromDataStream(dataStream)
        tableEnv.toRetractStream[Row](
            table.groupBy(("sid")).select('sid, 'sid.count as 'logNum)
        )
            .filter(_._1 == true)
            .print()

        tableEnv.execute("group")


    }

    /**
     * 创建动态表
     */
    def creatDynamicTable(dataStream: DataStream[StationLog]): Unit = {
        // 注册表
        tableEnv.registerDataStream("t_stream", dataStream)

        // 使用TableApi获取table对象
        val table = tableEnv.from("t_stream")
        table.printSchema()

        // 第二种 如果纯粹使用 TableAPI,推荐使用第二种, 如果使用SQL推荐使用第一种
        val table2 = tableEnv.fromDataStream(dataStream)
        table2.printSchema()

    }
}
